package com.whale.springboot.kafka.java.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Properties;

/**
 * 普普通通的消费者
 */
public class CustomConsumer_chongpingheng {

    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.179.131:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "JKH1");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); //自动提交时异步提交，丢数据&&重复数据
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        ArrayList<String> topics = new ArrayList<>();
        topics.add("test-more");

        // 一个消费者 看不出来效果，前后启动两次，观察控制台
        kafkaConsumer.subscribe(topics, new ConsumerRebalanceListener() {
            // 重平衡过程中， 消费者会被取消先前所分配的主题、分区
            // 取消之后，consumer底层就会调用下面的方法
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {

                System.out.println("我被取消了如下主题，分区：" + collection);
            }

            // 重平衡过程中，消费者会被重新分配新的主题、分区
            // 分配好了新的主题、分区后，会调用下面的方法
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                System.out.println("我又被分配了如下主题、分区：" + collection);
            }
        });

        // 拉取数据打印
        while (true) {
            kafkaConsumer.poll(Duration.ofSeconds(1));
        }
    }
}